package com.zk.storm;
//每日国家总点击量
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.HashMap;
import java.util.Map;

public class DailyCountrySumBolt extends BaseRichBolt {
    private OutputCollector collector;
    private Map<String, Long> countrySumMap;  // 键：date_key#country，值：总点击量

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.countrySumMap = new HashMap<>();
    }

    @Override
    public void execute(Tuple input) {
        try {
            String dateKey = input.getStringByField("date_key");
            String country = input.getStringByField("country");
            int hits = input.getIntegerByField("hits");

            String key = dateKey + "#" + country;  // 组合日期+国家作为唯一键
            countrySumMap.put(key, countrySumMap.getOrDefault(key, 0L) + hits);

            // 输出聚合结果（键、总点击量）
            collector.emit(new Values(key, countrySumMap.get(key)));
            collector.ack(input);
        } catch (Exception e) {
            collector.fail(input);
            throw new RuntimeException("国家总点击量聚合失败: " + e.getMessage());
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("agg_key", "sum_hits"));  // agg_key=date#country
    }
}